package com.gxkj.spark.alions.calculate

import java.util.Properties

import com.aliyun.openservices.ons.api._
import org.apache.spark.Logging
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.receiver.Receiver

//val consumerId = "CID-bigdata";
//val accessKey = "arzNQ1bmmyOSjHtt";
//val secretKey = "yFHjBg4HXVwBSyMkTrxQWcuinAi2o7";
//val topic = "bigdata";
//val key = "*";
class AliOnsReceiver (consumerId: String,accessKey:String,secretKey:String,topic:String, key: String)  extends Receiver[String](StorageLevel.MEMORY_AND_DISK_2) with Logging {
  /**
    * A custom receiver must extend this abstract class by implementing two methods
    * onStart(): Things to do to start receiving data.
    * onStop(): Things to do to stop receiving data.
    *
    */
  def onStart() {
    // Start the thread that receives data over a connection
    new Thread("aliOns Receiver") {
      override def run() { receive() }
    }.start()
  }

  def onStop() {
    // There is nothing much to do as the thread calling receive()
    // is designed to stop by itself if isStopped() returns false
  }
  private def receive(): Unit = {
    val properties: Properties = new Properties
    properties.put(PropertyKeyConst.ConsumerId, consumerId)
    properties.put(PropertyKeyConst.AccessKey, accessKey)
    properties.put(PropertyKeyConst.SecretKey, secretKey)
    var userInput: String = null
    try {
      val consumer: Consumer = ONSFactory.createConsumer(properties)
      consumer.subscribe(topic, key, new MessageListener() {
        def consume(message: Message, context: ConsumeContext): Action = {
          userInput = new String(message.getBody,"UTF-8");
          store(userInput)
          return Action.CommitMessage
        }
      })
      consumer.start

    } catch {
    case e: java.net.ConnectException =>
      // restart if could not connect to server
      restart("Error connecting to " + consumerId , e)
    case t: Throwable =>
      // restart if there is any other error
      restart("Error receiving data", t)
  }
  }
}
